package com.atguigu.gmall.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DwdTradeCartAdd
 * Description:
 *
 * @Create 2024/4/20 14:52
 */
public class DwdTradeCartAdd extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeCartAdd().start(10013, 4, Constant.TOPIC_DWD_TRADE_CART_ADD);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //TODO kafka读取topic_db，创建动态表
        readOdsDb(tableEnv, Constant.TOPIC_DWD_TRADE_CART_ADD);
        //TODO 过滤出加购行为  `table` = 'cart_info' and (`type` = 'insert' or (`type` = 'update' and old['sku_num'] is not null and data['sku_num'] > old['sku_num']))
        Table cartInfo = tableEnv.sqlQuery("select `data`['id'] id,\n" +
                "       `data`['user_id'] user_id,\n" +
                "       `data`['sku_id'] sku_id,\n" +
                "       if(`type` = 'insert',\n" +
                "          `data`['sku_num'],\n" +
                "          CAST((CAST(`data`['sku_num'] as int) - CAST(`old`['sku_num'] as int)) as string)) sku_num,\n" +
                "       ts\n" +
                "from topic_db\n" +
                "where `table` = 'cart_info'\n" +
                "  and (`type` = 'insert'\n" +
                "    or (`type` = 'update'\n" +
                "        and `old`['sku_num'] is not null\n" +
                "        and CAST(`data`['sku_num'] as int) > CAST(`old`['sku_num'] as int)\n" +
                "           )\n" +
                "    );");

        //cartInfo.execute().print();
        tableEnv.createTemporaryView("cart_info", cartInfo);

        //TODO 将过滤出来的加购数据写到kafka主题
        tableEnv.executeSql("CREATE TABLE " + Constant.TOPIC_DWD_TRADE_CART_ADD + "\n" +
                "(\n" +
                "    id string,\n" +
                "    user_id string,\n" +
                "    sku_id string,\n" +
                "    sku_num string,\n" +
                "    ts bigint,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_CART_ADD));

        tableEnv.executeSql("insert into " + Constant.TOPIC_DWD_TRADE_CART_ADD + " select * from cart_info");

    }
}
